package com.fwj.storm.say;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

/**
 * @author 傅文江
 * <p>
 * 时间： 2017-11-07 10:20
 * <p>
 * 描述：
 * <p>
 * 版权： Copyright © 2017 新道科技股份有限公司
 **/
public class ExclaimBasicTopo {
    public static void main(String[] args) throws Exception {
        //定义拓扑
        TopologyBuilder builder = new TopologyBuilder();
        //定义一个spout(喷口)，id为"spout"
        builder.setSpout("spout", new RandomSpout());

        //定义了一个id为"exclaim"的bolt(螺栓)，并且按照随机分组获得"spout(喷口)"发射的tuple(元组)
        //.shuffleGrouping 是指明Storm按照何种策略将tuple分配到后续的bolt去。
        //两个exectuor去执行两个exclaimBasicBolt的task
        builder.setBolt("exclaim", new ExclaimBasicBolt(),2).shuffleGrouping("spout");

        //定义了一个id为"print"的bolt(螺栓)，并且按照随机分组获得"exclaim”发射出来的tuple(元组)
        //3个executor去执行3个PrintBolt的task
        builder.setBolt("print", new PrintBolt(),3).shuffleGrouping("exclaim");

        Config conf = new Config();
        conf.setDebug(false);

        if (args != null && args.length > 0) {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("test", conf, builder.createTopology());
            Utils.sleep(3000);
            cluster.killTopology("test");
            cluster.shutdown();
        }
    }
}
